package m3;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.Scanner;

/**
 * @author Chenwx
 * @date 2021/5/14 9:27
 */
public class Producer {
    public static void main(String[] args) throws MQClientException {
        // 创建事务消息生产者
        TransactionMQProducer p = new TransactionMQProducer("group3");
        // 设置 name server
        p.setNamesrvAddr("192.168.64.141:9876");
        // 启动生产者
        p.start();

        // 设置事务消息的监听器
        // 1. 执行本地事务
        // 2. 处理事务状态回查
        p.setTransactionListener(new TransactionListener() {
            // 执行本地事务，在这里完成业务运算
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                if (Math.random() < 1) { // 100% 执行这个测试
                    // 由于网络中断，或者本地事务执行缓慢，超过1分钟，事务状态未知
                    return LocalTransactionState.UNKNOW;
                }

                // 两个参数： 事务消息， 业务数据
                String s = (String) arg;
                System.out.println("执行业务运算， data="+s);
                // 返回事务执行状态，把状态发送到 rocketmq 服务器
                if (Math.random() < 0.5) {
                    System.out.println("本地事务执行成功");
                    return LocalTransactionState.COMMIT_MESSAGE; //提交事务消息后，消息可以向消费者投递
                } else {
                    System.out.println("本地事务执行失败");
                    return LocalTransactionState.ROLLBACK_MESSAGE; //撤回消息
                }
                // return LocalTransactionState.UNKNOW; // 一般在代码中不会直接使用，而是由于网络中断，服务器不知道事务状态
            }

            // 处理 rocketmq 服务器的事务回查操作
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                System.out.println("服务器正在回查事务状态");
                return LocalTransactionState.UNKNOW;
            }
        });

        // 发送事务消息
        // 发送事务消息会触发一个监听器，执行本地事务
        while (true) {
            System.out.print("输入消息： ");
            String s = new Scanner(System.in).nextLine();
            Message msg = new Message("Topic3", s.getBytes());//消息数据的封装对象
            // 发送消息后，要执行本地事务，执行业务处理，需要向监听器传递业务数据
            p.sendMessageInTransaction(msg, "业务数据");
            System.out.println("事务消息已发送成功");
        }
    }
}

